package com.ce.source.analysis.mq;

import org.apache.rocketmq.client.consumer.DefaultMQPushConsumer;
import org.apache.rocketmq.client.consumer.listener.ConsumeConcurrentlyContext;
import org.apache.rocketmq.client.consumer.listener.ConsumeConcurrentlyStatus;
import org.apache.rocketmq.client.consumer.listener.MessageListenerConcurrently;
import org.apache.rocketmq.client.exception.MQClientException;
import org.apache.rocketmq.common.message.MessageExt;

import java.io.UnsupportedEncodingException;
import java.util.List;

/**
 * @author ：前世小和尚
 * @Description ：订单消费
 * @ClassName ：OrderConsumer
 * @date ：2018/3/11 12:28:28
 */
public class OrderConsumer {
    public static void main(String[] args) {
        DefaultMQPushConsumer consumer = new DefaultMQPushConsumer("gansu-order");
        try {
            consumer.subscribe("order", "*");
            consumer.setNamesrvAddr("192.168.199.245:9876");
            consumer.registerMessageListener(new MessageListenerConcurrently() {
                public ConsumeConcurrentlyStatus consumeMessage(List<MessageExt> msgs, ConsumeConcurrentlyContext context) {
//                    System.out.println("消息条数：" + msgs.size());
                    for (MessageExt ext : msgs) {
                        try {
                            String msg = new String(ext.getBody(), "utf-8");
                            System.out.println(ext.getMsgId() + "\t" + msg);
                            if (msg.equals("hello world 6")) {
                                System.out.println(msg + "\t 消息消费失败");
                                int a = 1 / 0;
                            }
                        } catch (Exception e) {
                            System.out.println(ext.getReconsumeTimes());
                            return ConsumeConcurrentlyStatus.RECONSUME_LATER;
                        }
                    }
                    return ConsumeConcurrentlyStatus.CONSUME_SUCCESS;
                }
            });
            consumer.start();
            System.out.println("消费者启动成功");
        } catch (Exception e) {
            e.printStackTrace();
        }
    }
}
